// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "threadpool.h"

#include <limits.h>
#include <pthread.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
#include <sched.h>

#include "common.h"

using namespace std;

ThreadPool::ThreadPool(int type_, int size_, int delay_) {
  // Initialize status variables.
  next_thread_id = type_ * 10000;
  num_running_threads = 0;

  // Initialize some parameters.
  type = type_;
  size = (size_ < 0 ? INT_MAX : size_);
  delay = delay_;
  
  // Initialize lock
  mutex_threads_to_run = new pthread_mutex_t;
  pthread_mutex_init(mutex_threads_to_run, NULL);
}

ThreadPool::~ThreadPool() {
}

int ThreadPool::GetNewThreadID() {
  next_thread_id = (next_thread_id % INT_MAX) + 1;
  return next_thread_id;
}

void ThreadPool::Run() {
  debug("Threadpool %d: Running.\n", type);
  while (true) {
    debug("Threadpool %d: Cleaning out threads marked as done.\n", type);
    // Clean out the finished threads.
    for (map<int, int>::iterator it = threads_status.begin(); 
         it != threads_status.end();
         it++) {
      if (it->second == kThreadStatusMarkAsDone) {
        // Thread is done, delete it from the list of running_threads.
        Thread* finished_thread = running_threads[it->first];
        delete finished_thread;
        // Also erase it from the map of running threads.
        running_threads.erase(it->first);
        SetStatus(it->first, kThreadStatusDone);
        num_running_threads--;
      }
    }

    debug("Threadpool %d: Attempt to run new threads.\n", type);
    // Next, see if we can run some new threads.
    while (num_running_threads < size && !threads_to_run.empty()) {
      // Pop the next thread to run. FIFO style.
      pthread_mutex_lock(mutex_threads_to_run);
      Thread* next_thread = threads_to_run.front().second;
      int next_thread_id = threads_to_run.front().first;
      threads_to_run.pop();
      pthread_mutex_unlock(mutex_threads_to_run);

      // Update the maps and statuses.
      running_threads[next_thread_id] = next_thread;
      SetStatus(next_thread_id, kThreadStatusRunning);

      // Run the thread!
      debug("Threadpool %d: Starting thread %d (type %d).\n",
            type, next_thread_id, next_thread->GetType());
      debug("Threadpool %d: Size of ToDo list %d.\n", type,
            static_cast<int>(threads_to_run.size()));
      next_thread->Start(next_thread_id);
      num_running_threads++;
    }

    sleep(delay);
    sched_yield();
  }
}

int ThreadPool::AddThread(Thread *thread) {
  int id = GetNewThreadID();
  
  pthread_mutex_lock(mutex_threads_to_run);
  threads_to_run.push(pair<int, Thread*>(id, thread));
  pthread_mutex_unlock(mutex_threads_to_run);
  threads_status[id] = kThreadStatusQueued;
  debug("Threadpool %d: Add thread %d of type %d.\n", type, id,
        thread->GetType());
  return id;
}

void ThreadPool::SetStatus(int id, int status) {
  // Only allow threads that are already in the status map to be updated, since
  // these are threads that are being run or have already been run.
  if (threads_status.find(id) != threads_status.end()) {
    threads_status[id] = status;
    debug("Threadpool %d: Thread %d changed status to %d.\n", type, id, status);
  }
}

int ThreadPool::GetStatus(int id) {
  if (threads_status.find(id) == threads_status.end()) {
    return kThreadStatusUndefined;
  } else {
    return threads_status[id];
  }
}

bool ThreadPool::HasJobsRunning() {
  debug("Threadpool %d: Running: %d, enqueued: %d.\n", type,
        num_running_threads, static_cast<int>(threads_to_run.size()));
  return (num_running_threads || !threads_to_run.empty());
}
